package com.lingshu.app.utils;

import com.lingshu.app.config.DefaultConfig;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class RabbitMQUtils {
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(DefaultConfig.getInstance().getMqHost());
        factory.setPort(DefaultConfig.getInstance().getMqPort());
        factory.setUsername(DefaultConfig.getInstance().getMqUsername());
        factory.setPassword(DefaultConfig.getInstance().getMqPassword());
        factory.setVirtualHost(DefaultConfig.getInstance().getMqVirtualHost());
        return factory.newConnection();
    }

    public static Channel getChannel(Connection connection) throws IOException {
        return connection.createChannel();
    }

    public static void receiveMessage(Channel channel, String exchangeName, androidx.core.util.Consumer<String> processor) throws IOException {
        String queueName = channel.queueDeclare().getQueue();
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT);
        channel.queueBind(queueName, exchangeName, "");
        Consumer callback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body);
                if (processor != null) {
                    processor.accept(message);
                }
            }
        };
        //channel.basicQos(0, 1, false);
        channel.basicConsume(queueName, true, callback);
    }
}
